package cn.cs.message.producer;

import java.io.IOException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.cs.message.ConnectionManage;
import cn.cs.message.Message;
import cn.cs.message.MessageClient;
import cn.cs.message.ReconnectHandler;
import cn.cs.message.support.MessageNameService;
import cn.cs.message.tempstorage.MemoryTempStorage;
import cn.cs.message.tempstorage.TempStorage;
import cn.cs.message.tempstorage.TempStorageEventListener;
import cn.cs.message.tempstorage.disk.DiskTempStorage;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BlockedListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.client.ReturnListener;
import com.rabbitmq.client.ShutdownSignalException;

/**消息生产者.调用sendMessage(...)发送消息
 * @version 1.0.0*/
public class MessageProducer extends MessageClient 
{
	private static final Logger logger=LoggerFactory.getLogger(MessageProducer.class);
	
	public void setThreadPoolNum(int threadPoolNum) {
		this.threadPoolNum = threadPoolNum;
	}
	
	public void setTempStorageEventListener(
			TempStorageEventListener tempStorageEventListener) {
		this.tempStorageEventListener = tempStorageEventListener;
	}
	//========================================================================================
	
	public boolean start() throws IOException, TimeoutException
	{
		messageNameService=new MessageNameService();
		messageNameService.setAtoB(direction);
		
		//重连机制初始化
		initReconnector();
		
		//启动线程池
		intiWoker();
		
		//初始化连接器
		if(!initConnector())
			return false;
		
		//初始化暂存器
		initTempStorage();
		
		return true;
	}
	
	/**##########################################重连机制############################################*/
	/***重连机制*/
	private ConcurrentHashMap<Thread, Channel> channelManager;
	/**重连机制初始化*/
	private void initReconnector()
	{
		channelManager=new ConcurrentHashMap<Thread, Channel>();
		reconnectHandler=new ReconnectHandler() 
		{
			@Override
			public void afterReconnect() {
				TempStorage oldTempStorage=tempStorage;
				tempStorage=tempStorage.newTempStorage();
				pushTempStorageMsg(oldTempStorage);
			}
		};
	}
	
	/**##############################线程池(接收发送消息) ###########################################*/
	/**线程池配置*/
	private ExecutorService threadPool;
	/**消息生产者的线程个数*/
	private int threadPoolNum;
	/**初始化工作线程*/
	private void intiWoker()
	{
		threadPool=new ThreadPoolExecutor(threadPoolNum, threadPoolNum,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new ThreadFactory(){
			@Override
			public Thread newThread(Runnable r) 
			{
				Thread thread=new Thread(r);
				Channel channel = connectManage.createChannel(new ReturnListener() {
					@Override
					public void handleReturn(int replyCode, String replyText,String exchange,
				            String routingKey,AMQP.BasicProperties properties, byte[] body) throws IOException {
						reminder.remind("原因："+replyText+" , exchange:"+exchange+":上routingKey"+routingKey+"的消息找不到队列，可能队列丢失，请马上立即修复丢失");
						tempStorage.add(getFromBytes(body));
					}
				});
				if(channel!=null)
					channelManager.put(thread, channel);
				
				return thread;
			}			
		}){
			 protected void afterExecute(Runnable r, Throwable t) {
				 super.afterExecute(r, t);
				 if(t!=null)
				 {
					 logger.error("Throwable",t);
					 //关闭，删除channel
					 Thread thread = Thread.currentThread();
					 Channel channel = channelManager.get(thread);
					 try {
						channel.close();
					} catch (IOException e) {
						e.printStackTrace();
					} catch (TimeoutException e) {
						e.printStackTrace();
					}finally{
						channelManager.remove(thread);
					}
				 }
			 }
		};
	}
	
	/**##########################################重连机制，可以重连的连接管理##########################################*/
	/***初始化连接管理器
	 * @throws TimeoutException 
	 * @throws IOException */
	private boolean initConnector() throws IOException, TimeoutException
	{
		try {
			connectManage = new ConnectionManage(connectUri,reconnectHandler,new BlockedListener() {
				@Override
				public void handleUnblocked() throws IOException {
					
				}
				@Override
				public void handleBlocked(String reason) throws IOException {
					reminder.remind(reason);
				}
			});
			logger.info("#########################MQ连接成功#####################################");
			return true;
		} catch (KeyManagementException e) {
			logger.error(e.getMessage(),e);
			e.printStackTrace();
		} catch (NoSuchAlgorithmException e) {
			logger.error(e.getMessage(),e);
			e.printStackTrace();
		} catch (URISyntaxException e) {
			logger.error(e.getMessage(),e);
			e.printStackTrace();
		}
		return false;
	}
	
	
	/**##########################################暂存消息，消息暂存器##########################################*/
	private String tempStorageType="disk";	//disk, memory
	private volatile TempStorage tempStorage;
	private TempStorageEventListener tempStorageEventListener=new TempStorageEventListener() {
		@Override
		public void onIncreaseMsg(long count) {
			reminder.remind("消息堆积："+count+"条");
		}
		@Override
		public void onError(String error) {
			reminder.remind(error);
		}
	};
	private void initTempStorage()
	{
		if(tempStorageType.equals("disk"))
			tempStorage=new DiskTempStorage();
		else
			tempStorage=new MemoryTempStorage();
		tempStorage.setListener(tempStorageEventListener);
		tempStorage.init();
		pushTempStorageMsg(tempStorage);
	}
	
	/**发送之前的暂存的消息。*/
	private void pushTempStorageMsg(TempStorage tempStorage)
	{
		if(!tempStorage.hasMsg())
			return;
		
		Channel channel;
		try {
			channel = connectManage.getConnection().createChannel();
			Iterator<Message> it=tempStorage.iterator();
			while(it.hasNext())
			{
				Message msg=it.next();
				try {
					sendToRmq(channel,msg);	
				} catch (IOException e) {
					logger.warn(e.getMessage(),e);
				}	
			}
			tempStorage.clear();
			channel.close();
		} catch (IOException e) {
			logger.warn(e.getMessage(),e);
		} catch (TimeoutException e) {
			logger.warn(e.getMessage(),e);
		}			
	}
	
	/**发送消息
	 * @throws IOException */
	private void sendToRmq(Channel channel,Message msg) throws IOException
	{
		String moduleCode=msg.getModuleCode();
		String exchange=messageNameService.getExchangeName(moduleCode);
		String routingKey=messageNameService.getRoutingKeyName(moduleCode);
		logger.debug("BEGIN:==================send message:exchange:"+exchange+";routingKey:"+routingKey+";" + msg);
		channel.basicPublish(exchange,routingKey,true,false,
				MessageProperties.PERSISTENT_TEXT_PLAIN,
				getBytes(msg));
	}
	
	
	/**发送消息
	 * @param msg 消息实体*/
	public void sendMessage(final Message msg)
	{
		if (this.connectManage == null || this.threadPool == null)
		{
			tempStorage.add(msg);
		}
		threadPool.execute(new Runnable() {
			@Override
			public void run() {
				if(!connectManage.isCanConnect())
				{
					tempStorage.add(msg);
					return;
				}
				
				Thread thread = Thread.currentThread();
				Channel channel = channelManager.get(thread);
				if (channel == null) 
				{
					channel = connectManage.createChannel();
					if (channel != null)
						channelManager.put(thread, channel);
					else 
					{
						tempStorage.add(msg);
						return;
					}
				}
				try {
					sendToRmq(channel,msg);	
				} catch (ShutdownSignalException e1) {
					channel = channelManager.get(thread);
					if(channel==null)
					{
						tempStorage.add(msg);
						return;
					}
					else if (!channel.isOpen()) 
					{
						channel = connectManage.createChannel();
						if (channel != null)
							channelManager.put(thread, channel);
						else
						{
							tempStorage.add(msg);
							return;
						}
					}
					try {
						sendToRmq(channel,msg);	
					} catch (IOException e) {
						logger.warn(e.getMessage(),e);
						tempStorage.add(msg);
					}
				} catch (IOException e) {
					logger.warn(e.getMessage(),e);
					tempStorage.add(msg);
				}
			}
		});
	}
	
	public void stop() throws IOException, TimeoutException 
	{
		Collection<Channel> channels=channelManager.values();
		for(Channel temp:channels)
			temp.close();
		
		connectManage.shutdown();
		
		if(threadPool!=null)
			threadPool.shutdown();
	}


	public int getThreadPoolNum() {
		return threadPoolNum;
	}

}
